package com.qf.cx.simplenms.subscriber.impl;

import java.util.Observable;
import java.util.Observer;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.qf.cx.simplenms.common.utils.SetUtils;
import com.qf.cx.simplenms.report.model.Report;
import com.qf.cx.simplenms.subscriber.api.Subscriber;
import com.qf.cx.simplenms.subscriber.model.SubscriberModel;
import com.qf.cx.simplenms.subscriber.spi.SubscriberSpi;
import com.qf.cx.simplenms.subscriber.type.FileSubscriberOutputChannel;
import com.qf.cx.simplenms.subscriber.type.HTTPSubscriberOutputChannel;
import com.qf.cx.simplenms.subscriber.type.SubscriberOutputChannel;
import com.qf.cx.simplenms.subscriber.type.TCPSubscriberOutputChannel;
import com.qf.cx.simplenms.subscriber.type.exception.InvalidURIException;
import com.qf.cx.simplenms.topic.api.Topic;

public class SubscriberImpl implements SubscriberSpi, Subscriber, Observer, Runnable {

	private Logger logger = LoggerFactory.getLogger(this.getClass());
	/** http uri prefix */
	private static final String HTTP_PREFIX = "http";
	/** tcp uri prefix */
	private static final String TCP_PREFIX = "tcp";
	/** file uri prefix */
	private static final String FILE_PREFIX = "file";
	private final SubscriberModel subscriberModel;
	private Thread reportSender;
	private SubscriberOutputChannel subscriberOutputChannel;
	private final MaxSizeQueue<Report> queue;

	private final AtomicBoolean started = new AtomicBoolean(false);

	public SubscriberImpl(int maxQueueSize, SubscriberModel subscriberModel) throws InvalidURIException {
		this.queue = new MaxSizeQueue<>(maxQueueSize);
		this.subscriberModel = subscriberModel;
		String notificationURI = this.subscriberModel.getUrl();
		String protocol;
		try {
			String[] parts = notificationURI.split(":");
			protocol = parts[0];
		} catch (Exception ex) {
			throw new InvalidURIException("illegal notification URI: " + notificationURI, ex);
		}

		if (HTTP_PREFIX.equalsIgnoreCase(protocol)) {
			logger.debug("using http subscriber output channel: " + notificationURI);
			subscriberOutputChannel = new HTTPSubscriberOutputChannel(notificationURI);
		} else if (TCP_PREFIX.equalsIgnoreCase(protocol)) {
			logger.debug("using tcp subscriber output channel: " + notificationURI);
			subscriberOutputChannel = new TCPSubscriberOutputChannel(notificationURI);
		} else if (FILE_PREFIX.equalsIgnoreCase(protocol)) {
			logger.debug("using file subscriber output channel: " + notificationURI);
			subscriberOutputChannel = new FileSubscriberOutputChannel(notificationURI);
		} else {
			// invalid url
			throw new InvalidURIException("Invalid protocol.");
		}
	}

	@Override
	public void startIfNecessary() {
		if (!started.compareAndSet(false, true)) {
			logger.debug("subscriber '{}' already started", getUrl());
			return;
		}

		logger.info("subscriber '{}' starting", getUrl());
		reportSender = new Thread(this, getUrl() + "_ReportsSender");
		reportSender.start();
		getTopics().forEach(t -> {
			startTopic(t);
		});
		logger.info("subscriber '{}' started", getUrl());
	}

	@Override
	public void stop() {
		if (!started.compareAndSet(true, false)) {
			return;
		}
		stop0();
	}

	@Override
	public void forceStop() {
		stop0();
	}

	private void stop0() {
		logger.info("subscriber '{}' stopping", getUrl());
		if (reportSender != null) {
			reportSender.interrupt();
		}
		getTopics().forEach(t -> {
			stopTopic(t);
		});
		subscriberOutputChannel.close();
		logger.info("subscriber '{}' stopped", getUrl());
	}

	@Override
	public void run() {
		for (;;) {
			Report report = null;
			try {
				TimeUnit.MILLISECONDS.sleep(1);
				report = queue.take();
			} catch (InterruptedException e) {
				logger.warn("'{}' stopped", Thread.currentThread().getName());
				return;
			}
			subscriberOutputChannel.push(report);
		}
	}

	@Override
	public void update(Observable o, Object arg) {

		if (!started.get()) {
			return;
		}
		if (!(arg instanceof Report)) {
			logger.error("can not handle type {}", arg.getClass());
			return;
		}

		Report report = (Report) arg;
		queue.add(report);
	}

	private String getUrl() {
		return subscriberModel.getUrl();
	}

	private Set<Topic> getTopics() {
		return subscriberModel.getTopics();
	}

	@Override
	public void updateSubscriberModel(SubscriberModel t) {
		synchronized (this.subscriberModel) {
			Set<Topic> news = t.getTopics();
			Set<Topic> olds = this.subscriberModel.getTopics();
			Set<Topic> dels = SetUtils.getComplementary(olds, news);
			deleteTopics(dels);

			Set<Topic> adds = SetUtils.getComplementary(news, olds);
			addTopics(adds);
		}
	}

	private void addTopics(Set<Topic> adds) {
		if (adds.isEmpty()) {
			return;
		}
		logger.info("subscriber '{}' add topics '{}'", getUrl(), adds);
		if (started.get()) {
			adds.forEach(t -> {
				startTopic(t);
			});
		}
		subscriberModel.addTopics(adds);
	}

	private void deleteTopics(Set<Topic> deletes) {
		if (deletes.isEmpty()) {
			return;
		}
		logger.info("subscriber '{}' delete topics '{}'", getUrl(), deletes);
		if (started.get()) {
			deletes.forEach(t -> {
				stopTopic(t);
			});
		}
		subscriberModel.deleteTopics(deletes);
	}

	private void startTopic(Topic t) {
		t.start(this);
	}

	private void stopTopic(Topic t) {
		t.stop(this);
	}

	@Override
	public String toString() {
		return "SubscriberImpl [model=" + subscriberModel + ", started=" + started + "]";
	}

	@Override
	public String dump() {
		StringBuilder sb = new StringBuilder();
		sb.append("started=").append(started);
		sb.append(", url=").append(subscriberModel.getUrl());
		sb.append(", topics:[");
		for (Topic e : subscriberModel.getTopics()) {
			sb.append("[");
			sb.append(e.getName());
			sb.append("]");
		}
		sb.append("]");

		return sb.toString();
	}

}
